package com.moli.iov.common.hbase.autoconfigure;

import com.google.common.base.Joiner;
import com.moli.iov.common.hbase.service.impl.HBaseServiceImpl;
import com.moli.iov.common.hbase.enums.ThreadPoolExecutorPolicy;
import com.moli.iov.common.hbase.properties.HBaseProperties;
import com.moli.iov.common.hbase.service.IHbaseService;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;
import org.apache.hadoop.hbase.exceptions.HBaseException;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.io.IOException;
import java.util.List;
import java.util.concurrent.*;

import static com.moli.iov.common.hbase.constants.HBaseConstant.*;
import static org.apache.hadoop.hbase.HConstants.*;

/**
 *
 * {@link EnableAutoConfiguration Auto-configuration} for iov HBase connection.
 *
 * @author OAK
 * @version  3.0.1
 * @since 2019/12/13 10:12:00 PM.
 *
 */
@Configuration
@ConditionalOnClass(Connection.class)
@EnableConfigurationProperties(HBaseProperties.class)
public class AutoConfiguration {

    @Bean
    @ConditionalOnMissingBean
    public Connection connection(HBaseProperties properties) throws HBaseException {
        if(properties == null){
            throw new HBaseException("HBase not found configuration properties.");
        }
        HBaseProperties.ZookeeperProperties zookeeperProperties = properties.getZookeeper();
        if(zookeeperProperties == null || zookeeperProperties.getQuorum() == null){
            throw new HBaseException("HBase zookeeper quorum not found.");
        }
        org.apache.hadoop.conf.Configuration configuration = HBaseConfiguration.create();
        List<String> quorums = zookeeperProperties.getQuorum();
        configuration.set(ZOOKEEPER_QUORUM, Joiner.on(CONCAT_SEPARATOR).join(quorums));
        HBaseProperties.ZookeeperProperties.ZookeeperPropertyProperties zookeeperPropertyProperties = zookeeperProperties.getProperty();
        if(zookeeperPropertyProperties == null || zookeeperPropertyProperties.getClientPort() == null){
            configuration.setInt(ZOOKEEPER_CLIENT_PORT, DEFAULT_ZOOKEPER_CLIENT_PORT);
        }else{
            Integer clientPort = zookeeperPropertyProperties.getClientPort();
            configuration.setInt(ZOOKEEPER_CLIENT_PORT, clientPort);
        }

        HBaseProperties.RpcProperties rpcProperties = properties.getRpc();
        if(rpcProperties == null || rpcProperties.getTimeout() == null){
            configuration.setInt(HBASE_RPC_TIMEOUT_KEY, DEFAULT_HBASE_RPC_TIMEOUT);
        }else{
            Integer timeout = rpcProperties.getTimeout();
            configuration.setInt(HBASE_RPC_TIMEOUT_KEY, timeout);
        }

        HBaseProperties.ClientProperties client = properties.getClient();
        if(client != null){
            HBaseProperties.ClientProperties.ClientIpcProperties clientIpcProperties = client.getIpc();
            if(clientIpcProperties != null && clientIpcProperties.getPool() != null){
                Integer size = clientIpcProperties.getPool().getSize();
                String type = clientIpcProperties.getPool().getType();
                if(size != null){
                    configuration.setInt(HBASE_CLIENT_IPC_POOL_SIZE, size);
                }
                if(type != null){
                    configuration.set(HBASE_CLIENT_IPC_POOL_TYPE, type);
                }
            }

          HBaseProperties.ClientProperties.ClientOperationProperties clientOperationProperties = client.getOperation();
            if(clientOperationProperties != null && clientOperationProperties.getTimeout() != null){
                Integer operationTimeOut = clientOperationProperties.getTimeout();
                configuration.setInt(HBASE_CLIENT_OPERATION_TIMEOUT, operationTimeOut);
            }else{
                configuration.setInt(HBASE_CLIENT_OPERATION_TIMEOUT, DEFAULT_HBASE_CLIENT_OPERATION_TIMEOUT);
            }

            HBaseProperties.ClientProperties.ClientScannerProperties clientScannerProperties = client.getScanner();

            if(clientScannerProperties != null && clientScannerProperties.getTimeout() != null){
                HBaseProperties.ClientProperties.ClientScannerProperties.ClientScannerTimeoutProperties
                        clientScannerTimeoutProperties = clientScannerProperties.getTimeout();
                if(clientScannerTimeoutProperties != null && clientScannerTimeoutProperties.getPeriod() != null){
                    Integer clientScannerTimeoutPeriod = clientScannerTimeoutProperties.getPeriod();
                    configuration.setInt(HBASE_CLIENT_OPERATION_TIMEOUT, clientScannerTimeoutPeriod);
                }else{
                    configuration.setInt(HBASE_CLIENT_SCANNER_TIMEOUT_PERIOD, DEFAULT_HBASE_CLIENT_SCANNER_TIMEOUT_PERIOD);
                }
            }else{
                configuration.setInt(HBASE_CLIENT_SCANNER_TIMEOUT_PERIOD, DEFAULT_HBASE_CLIENT_SCANNER_TIMEOUT_PERIOD);
            }
        }else{
            configuration.setInt(HBASE_CLIENT_IPC_POOL_SIZE, DEFAULT_HBASE_CLIENT_IPC_POOL_SIZE);
            configuration.set(HBASE_CLIENT_IPC_POOL_TYPE, DEFAULT_HBASE_CLIENT_IPC_POOL_TYPE);
            configuration.setInt(HBASE_CLIENT_SCANNER_TIMEOUT_PERIOD, DEFAULT_HBASE_CLIENT_SCANNER_TIMEOUT_PERIOD);
        }

        Boolean enableAsync = properties.getEnableAsync(),
            enableThreadPool = properties.getEnableThreadPool();

        if(enableAsync != null && enableAsync){
            //This version does not support asynchronous.
        }
        Connection connection = null;
        try {
            if (enableThreadPool != null && enableThreadPool) {
                HBaseProperties.ThreadPoolProperties threadPoolProperties = properties.getPool();
                Integer corePoolSize = threadPoolProperties.getCorePoolSize(),
                        maxPoolSize = threadPoolProperties.getMaxPoolSize(),
                        keepAliveSeconds = threadPoolProperties.getKeepAliveSeconds(),
                        queueCapacity = threadPoolProperties.getQueueCapacity();
                String executorPolicy = threadPoolProperties.getExecutorPolicy();
                ExecutorService executorService = new ThreadPoolExecutor(corePoolSize, maxPoolSize, keepAliveSeconds, TimeUnit.SECONDS, new LinkedBlockingQueue<>(queueCapacity), ThreadPoolExecutorPolicy.lookupForDisplayName(executorPolicy));
                connection = ConnectionFactory.createConnection(configuration, executorService);
            } else {
                connection = ConnectionFactory.createConnection(configuration);
            }
        }catch (IOException e){
            throw new HBaseException("Via factory connection to HBase occur cause", e);
        }
        return connection;
    }

    @Bean
    @ConditionalOnMissingBean
    public IHbaseService iHbaseService(Connection connection) throws HBaseException {
        return new HBaseServiceImpl(connection);
    }

}
